package node

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/libp2p/go-libp2p"
	circuit "github.com/libp2p/go-libp2p-circuit"
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/metrics"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-core/peerstore"
	"github.com/libp2p/go-libp2p-core/protocol"
	"github.com/multiformats/go-multiaddr"
	maddr "github.com/multiformats/go-multiaddr"
	logger "hundsun.com/hsl/hschain/common/log"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/p2p/common"
	"hundsun.com/hsl/hschain/p2p/common/constant"
	"hundsun.com/hsl/hschain/p2p/common/utils"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node/blacklist"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node/discovery"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node/p2pdb"
	protosp2p "hundsun.com/hsl/hschain/protos/p2p"

	hslcrypto "hundsun.com/hsl/hschain/common/crypto"

	pbp2p "hundsun.com/hsl/hschain/protos/p2p"
	dbm "hundsun.com/hsl/hschain/store/db"
)

//Node 节点
type Node struct {
	logging.Log
	conf *common.P2pConfig
	ctx  context.Context
	Host host.Host

	PeerInfoMgr *PeerInfoManager     // 用于记录当前连接节点的信息(如其中包括最新区块头信息)
	Discovery   *discovery.Discovery // 用于p2p节点发现
	ConnMgr     *ConnManager         // 连接管理
	DbHandler   *p2pdb.DbHandler
	Manager     *blacklist.PeerManager
}

//Init 节点初始化
func Init(ctx context.Context, conf *common.P2pConfig, logger logging.Log, db dbm.Database) (*Node, error) {
	bandwidth := metrics.NewBandwidthCounter()
	blacklistOpts, err := genBlackListOption(conf)
	if err != nil {
		logger.Error("genBlackListOption error!", "error", err.Error())
		return nil, err
	}
	options, muAddr, err := genHostOption(conf)
	if err != nil {
		logger.Error("genHostOption error!", "error", err.Error())
		return nil, err
	}

	dbh := p2pdb.NewDbHandler(db)

	ht, err := libp2p.New(ctx, options...)
	if err != nil {
		panic(err)
	}
	logger.Infof("p2p started, peerID:%v", ht.ID())

	ht.Peerstore().AddAddr(ht.ID(), muAddr, peerstore.PermanentAddrTTL)

	manager, err := blacklist.NewManager(dbh, logger, blacklistOpts...)
	if err != nil {
		return nil, err
	}

	n := &Node{
		Log:       logger,
		conf:      conf,
		ctx:       ctx,
		Host:      ht,
		DbHandler: dbh,
		Manager:   manager,
	}
	var peersInfo []peer.AddrInfo
	//todo 目前只支持单个初始化节点，sdk能生成peerid后可支持多个初始化节
	if len(conf.BootstrapPeers) > 1 {
		return nil, errors.New("more then one BootstrapPeers not supported")
	}
	if len(conf.BootstrapPeers) == 1 {
		add, err := maddr.NewMultiaddr(conf.BootstrapPeers[0])
		if err != nil {
			logger.Error("peer address not illegal, address:%s", conf.BootstrapPeers[0])
			return nil, err
		}
		peerInfo, err := peer.AddrInfoFromP2pAddr(add)
		if err != nil {
			logger.Error("init bootstrap peer failed, address illegal, %v", add)
			return nil, err
		}
		peersInfo = append(peersInfo, *peerInfo)
	}
	n.Discovery = discovery.NewDiscovery(ctx, ht, peersInfo, conf, logger)
	n.ConnMgr = NewConnManager(ctx, ht, n.Discovery.RoutingTable(), bandwidth, conf, logger)
	n.PeerInfoMgr = NewPeerInfoManager(ctx, ht)
	return n, nil
}

//Start 节点启动
func (n *Node) Start() error {
	n.Discovery.Start()
	go n.ConnMgr.Start()
	go n.PeerInfoMgr.Start()
	return nil
}

//Close 节点关闭
func (n *Node) Close() {
	if n.Discovery != nil {
		n.Discovery.Close()
	}
	if n.Host != nil {
		n.Host.Close()
	}
}

//SendMessage 发送信息
func (n *Node) SendMessage(msg proto.Message, ptl protocol.ID, pr peer.ID) error {
	stream, err := n.Host.NewStream(n.ctx, pr, ptl)
	if err != nil {
		n.Log.Error("sendPeer", "id", pr, "NewStreamErr", err)
		return err
	}
	err = WriteStream(msg, stream)
	if err != nil {
		n.Log.Error("sendPeer", "id", pr, "WriteStream err", err)
		return err
	}
	CloseStream(stream)
	return nil
}

func genBlackListOption(conf *common.P2pConfig) ([]blacklist.Option, error) {
	var blacklistOpts []blacklist.Option
	if hslcrypto.EnableAuthVerify() {
		blacklistOpts = append(blacklistOpts, blacklist.EnableTemporary())
	}
	if conf.ProtectPeers == nil {
		return blacklistOpts, nil
	}
	var peerItems []peer.AddrInfo
	for _, pp := range conf.ProtectPeers {
		p, err := utils.GenAddrInfo(pp)
		if err != nil {
			return blacklistOpts, err
		}
		peerItems = append(peerItems, *p)
	}
	blacklistOpts = append(blacklistOpts, blacklist.AddSystemWhite(peerItems))
	return blacklistOpts, nil
}

// genHostOption 生成 host option
func genHostOption(cfg *common.P2pConfig) ([]libp2p.Option, multiaddr.Multiaddr, error) {
	muAddr, err := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/0.0.0.0/tcp/%s", cfg.P2pPort))
	if err != nil {
		return nil, muAddr, err
	}

	var blacklistOpts []blacklist.Option
	if hslcrypto.EnableAuthVerify() {
		blacklistOpts = append(blacklistOpts, blacklist.EnableTemporary())
	}

	//读取peerID私钥
	privateKey, err := utils.GetPeerIDPrivateKey(cfg.PeerIDPriKeyPath)
	if err != nil {
		panic(err)
	}

	opts := []libp2p.Option{
		libp2p.ListenAddrs(muAddr),
		libp2p.EnableRelay(circuit.OptHop), // TODO 后续做成可以使能配置
		libp2p.Identity(privateKey),
	}

	if cfg.EnableIpv6 {
		muAddr, _ = multiaddr.NewMultiaddr(fmt.Sprintf("/ip6/::/tcp/%s", cfg.P2pPort))
		opts = append(opts, libp2p.ListenAddrs(muAddr))
	}

	if cfg.EnableNat {
		opts = append(opts, libp2p.NATPortMap())
	}
	return opts, muAddr, nil
}

//NewP2pMessage 新建p2p消息
func (n *Node) NewP2pMessage(version, chainID string, msgTy int32, payload []byte, reserved []byte) *pbp2p.P2PMsg {
	msg := &pbp2p.P2PMsg{
		Version:   version,
		ChainId:   chainID,
		MsgType:   msgTy,
		Timestamp: time.Now().UnixNano(),
		Payload:   payload,
		Reserved:  reserved,
		PeerId:    utils.PeerIDEncode(n.Host.ID()),
	}
	return msg
}

//NewConsensusP2pMessage 新建共识消息
func (n *Node) NewConsensusP2pMessage(version, chainID string, msgTy int32, targetPeerID string, targetMod string, payload []byte, reserved []byte) *pbp2p.P2PMsg {
	msg := &pbp2p.P2PMsg{
		Version:      version,
		ChainId:      chainID,
		MsgType:      msgTy,
		TargetMod:    targetMod,
		Timestamp:    time.Now().UnixNano(),
		Payload:      payload,
		Reserved:     reserved,
		PeerId:       utils.PeerIDEncode(n.Host.ID()),
		TargetPeerId: targetPeerID,
	}
	return msg
}

//AsyncMsg p2p异步消息
func (n *Node) AsyncMsg(data *protosp2p.P2PMsg, pid peer.ID, msgType protocol.ID) error {
	if n.Manager.Contains(pid) {
		return nil
	}
	stream, err := n.Host.NewStream(n.ctx, pid, msgType)
	if err != nil {
		return err
	}
	err = WriteStream(data, stream)
	defer CloseStream(stream)
	if err != nil {
		return err
	}
	return nil
}

//OriSyncMsg 不校验黑名单
func (n *Node) OriSyncMsg(req *protosp2p.P2PMsg, pid peer.ID, msgType protocol.ID) (*protosp2p.P2PMsg, error) {
	var resp = &protosp2p.P2PMsg{}
	req.PeerId = utils.PeerIDEncode(n.Host.ID())
	req.TargetPeerId = utils.PeerIDEncode(pid)
	ctx, cancel := context.WithTimeout(n.ctx, time.Millisecond*constant.P2pMsgTimeout)
	defer cancel()
	err := pid.Validate()
	if err != nil {
		return nil, err
	}
	stream, err := n.Host.NewStream(ctx, pid, msgType)
	if err != nil {
		return nil, err
	}
	defer CloseStream(stream)
	err = WriteStream(req, stream)
	if err != nil {
		return nil, err
	}

	done := make(chan error)
	go func() {
		done <- ReadStream(resp, stream)
	}()

	select {
	case <-done:
		return resp, nil
	case <-time.After(time.Millisecond * constant.P2pMsgTimeout):
		cancel()
		return nil, errors.New("p2p sync message timeout")
	}
}

//SyncMsg p2p同步消息
func (n *Node) SyncMsg(req *protosp2p.P2PMsg, pid peer.ID, msgType protocol.ID) (*protosp2p.P2PMsg, error) {
	if n.Manager.Contains(pid) {
		return nil, nil
	}
	return n.OriSyncMsg(req, pid, msgType)
}

//OperationSync p2p同步消息回复
type OperationSync func(msg *pbp2p.P2PMsg) *pbp2p.P2PMsg

//OperationAsync p2p异步消息处理
type OperationAsync func(msg *pbp2p.P2PMsg)

//OriSyncMsgResp 不校验黑名单
func (n *Node) OriSyncMsgResp(operation OperationSync) func(network.Stream) {
	//todo 两节点peerID重复校验
	return func(stream network.Stream) {
		msg := &protosp2p.P2PMsg{}
		if err := ReadStream(msg, stream); err != nil {
			n.Log.Errorf("p2p read stream failed, reason:%v", err)
			return
		}

		result := operation(msg)
		if err := WriteStream(result, stream); err != nil {
			n.Log.Errorf("p2p write stream failed, reason:%v", err)
			return
		}
	}
}

//SyncMsgResp 同步消息回复
func (n *Node) SyncMsgResp(operation OperationSync) func(network.Stream) {
	//todo 两节点peerID重复校验
	return func(stream network.Stream) {
		if n.Manager.Contains(stream.Conn().RemotePeer()) {
			CloseStream(stream)
			return
		}
		msg := &protosp2p.P2PMsg{}
		if err := ReadStream(msg, stream); err != nil {
			n.Log.Errorf("p2p read stream failed, reason:%v", err)
			return
		}

		result := operation(msg)
		if err := WriteStream(result, stream); err != nil {
			n.Log.Errorf("p2p write stream failed, reason:%v", err)
			return
		}
	}
}

//AsyncMsgResp 异步消息响应
func (n *Node) AsyncMsgResp(operation OperationAsync) func(network.Stream) {
	return func(stream network.Stream) {
		defer CloseStream(stream)
		if n.Manager.Contains(stream.Conn().RemotePeer()) {
			return
		}
		msg := &protosp2p.P2PMsg{}
		if err := ReadStream(msg, stream); err != nil {
			n.Log.Errorf("p2p read stream failed, reason:%v", err)
		}
		operation(msg)
	}
}

//AuthVerify 权限校验
func (n *Node) AuthVerify(certBuf []byte) bool {
	if !hslcrypto.EnableAuthVerify() {
		logger.Infof("check signature disabled")
		return true
	}
	//验证证书
	caBuf, err := hslcrypto.GetCaCertFile()
	if err != nil {
		n.Log.Errorf("get ca cert file failed, reason:%v", err)
		return false
	}

	//验证证书ip
	certificate, err := hslcrypto.CertFromBytes(certBuf)
	if err != nil || certificate == nil {
		n.Log.Errorf("decode cert file failed, reason:%v", err)
		return false
	}
	//不校验ip，因为监听的ip和证书中的ip可能不一致
	/*logger.Infof("check signature ipCert:%s, ip:%s", certificate.Subject.CommonName, ip)
	if hslcrypto.EnablePrivateCa() && certificate.Subject.CommonName != ip {
		return false, fmt.Errorf("ip not match")
	}*/
	//验证子证书由父证书签发
	bol, err := hslcrypto.CheckSignatureFrom(caBuf, certBuf)
	if err != nil {
		n.Log.Errorf("check cert failed, reason:%v", err)
	}
	return bol
}
